package cars;

import java.util.HashMap;
import java.util.Set;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class SexwithCar {

	public static void main(String[] args) throws Exception {
		if(args.length!=2){
			System.err.println("user path err!!");
			System.exit(-1);
		}
		@SuppressWarnings("deprecation")
		Job job=new Job(new Configuration(), "car");
		job.setJarByClass(SexwithCar.class);
		
		FileInputFormat.addInputPath(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job,new Path(args[1]));
		
		job.setMapperClass(scmap.class);
		job.setReducerClass(screduce.class);
		
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);
		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(DoubleWritable.class);
		job.waitForCompletion(true);
	}
	public static class scmap extends Mapper<LongWritable,Text,Text,IntWritable>{
		protected void map(LongWritable key, Text value, org.apache.hadoop.mapreduce.Mapper<LongWritable,Text,Text,IntWritable>.Context context) throws java.io.IOException ,InterruptedException {
		String[] lines = value.toString().split(",");
		if(lines.length>38 && lines[38]!=null && lines[38]!=""){
			String sex=lines[38].trim();
			context.write(new Text(sex), new IntWritable(1));
			//Map的结果：<man,1>....
		}
		};
	}
	/*map阶段产生的需要shuffle处理
	 * */
	public static class screduce extends Reducer<Text,IntWritable,Text,DoubleWritable>{
		double total=0;
		HashMap<String,Integer> map=new HashMap<String,Integer>();
 		protected void reduce(Text k2, java.lang.Iterable<IntWritable> values, org.apache.hadoop.mapreduce.Reducer<Text,IntWritable,Text,DoubleWritable>.Context context) throws java.io.IOException ,InterruptedException {
			int sum=0;//存放男女各自的数量
			for (IntWritable count : values) {
				sum+=count.get();
			}
			total+=sum;//计算出男女总数
			//分别将性别和数量存放到集合中
			map.put(k2.toString(),sum);
			
		};
		//reduce 函数处理完毕后才调用cleanUp函数
		protected void cleanup(org.apache.hadoop.mapreduce.Reducer<Text,IntWritable,Text,DoubleWritable>.Context context) throws java.io.IOException ,InterruptedException {
			Set<String> keySet = map.keySet();
			for (String sex : keySet) {
				double amount=map.get(sex);//通过key分别获取与之对应的value值，vlaue就是男女各自的数量
				double persent=amount/total;
				context.write(new Text("统计买车的"+sex+"比例"),new DoubleWritable(persent));
			}
		};
	}
}
